package com.clw.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.clw.KafkaUtils;
import com.clw.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.*;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;


/**
 * 求跳出率
 */
public class JumpDetail {

    public static void main(String[] args) throws Exception {

        // 创建流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 获取kafka 的流
        FlinkKafkaConsumer<String> consumer = MyKafkaUtil.getKafkaConsumer("ods_base_log", "safsfsd");


        DataStreamSource<String> logStream = env.addSource(consumer);


        // 过滤  进行消费  包含“common”
        SingleOutputStreamOperator<JSONObject> mapStream = logStream.map(x -> {
            return JSON.parseObject(x);
        }).filter(x -> {
            return x.containsKey("page") && x.containsKey("common");

        });


        // 设置水位线
        SingleOutputStreamOperator<JSONObject> ts = mapStream.assignTimestampsAndWatermarks(WatermarkStrategy.
                <JSONObject>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject jsonObject, long recordTimestamp) {
                return jsonObject.getLong("ts");
            }
        }));

        // 根据设备id分组
        KeyedStream<JSONObject, Object> streamByMid = ts.keyBy(x -> {
            return x.getJSONObject("common").getString("mid");
        });


        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("GoIn").where(
                new SimpleCondition<JSONObject>() {
                    @Override   // 条件1 ：进入的第一个页面
                    public boolean filter(JSONObject jsonObj) throws Exception {
                        String lastPageId = jsonObj.getJSONObject("page").getString("last_page_id");
                        //  System.out.println("first in :" + lastPageId);
                        if (lastPageId == null || lastPageId.length() == 0) {
                            return true;
                        }
                        return false;
                    }
                }
        ).next("next").where(
                new SimpleCondition<JSONObject>() {
                    @Override  //条件2： 在10秒时间范围内必须有第二个页面
                    public boolean filter(JSONObject jsonObj) throws Exception {
                        String pageId = jsonObj.getJSONObject("page").getString("page_id");
                        // System.out.println("next:" + pageId);
                        if (pageId != null && pageId.length() > 0) {
                            return true;
                        }
                        return false;
                    }
                }
        ).within(Time.milliseconds(10000));


        // 将规则应用到流上
        PatternStream<JSONObject> patternStream = CEP.pattern(streamByMid, pattern);

        // 定义超时事件
        //TODO 6.提取匹配上的和超时事件
        final OutputTag<String> timeoutTag = new OutputTag<String>("timeout") {
        };
        SingleOutputStreamOperator<String> filteredStream = patternStream.flatSelect(
                timeoutTag,
                new PatternFlatTimeoutFunction<JSONObject, String>() {
                    @Override
                    public void timeout(Map<String, List<JSONObject>> pattern, long timeoutTimestamp, Collector<String> out) throws Exception {
                        List<JSONObject> objectList = pattern.get("GoIn");
                        //这里进入out的数据都被timeoutTag标记
                        for (JSONObject jsonObject : objectList) {
                            out.collect(jsonObject.toJSONString());
                        }
                    }
                },
                new PatternFlatSelectFunction<JSONObject, String>() {
                    @Override
                    public void flatSelect(Map<String, List<JSONObject>> pattern, Collector<String> out) throws Exception {
                        //因为不超时的事件不提取，所以这里不写代码
                    }
                });


        DataStream<String> jumpDstream = filteredStream.getSideOutput(timeoutTag);

        //  filteredStream.print();

        jumpDstream.print();

        // jumpDstream.print();
        jumpDstream.addSink(KafkaUtils.createProduer("dwm_user_jump_detail"));

        env.execute();

    }

}
